package delay;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.List;

/**
 * delay包,延迟消息
 * 消费者不需要做处理，由broker来实现延迟消息。
 * @author: BacHe
 * @email: 1218585258@qq.com
 * @Date: 2020/12/9 22:45
 */
public class Consumer {
    public static void main(String[] args) throws MQClientException {
        //1.创建消费者consumer，指定消费者组
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group5");
        //2.指定nameserver地址
        consumer.setNamesrvAddr("192.168.0.88:9876");
        //3.订阅主题Topic 和 tag
        consumer.subscribe("delay","tag1");
        consumer.setMessageModel(MessageModel.CLUSTERING);
        //4.设置回调函数，处理消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                list.forEach(messageExt -> {
                    String time = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
                    System.out.println(time+",consumeMessage:" + new String(messageExt.getBody()));
                });
                //返回消费成功
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        //5.启动消费者consumer
        consumer.start();

// 控制台输出如下,设置10秒延迟已经实现，误差在1秒之内。

//        2020-12-09 23:09:51,consumeMessage:rocketmq 发送延迟消息 0,time:2020-12-09 23:09:40
//        2020-12-09 23:09:52,consumeMessage:rocketmq 发送延迟消息 1,time:2020-12-09 23:09:42
//        2020-12-09 23:09:53,consumeMessage:rocketmq 发送延迟消息 2,time:2020-12-09 23:09:43
//        2020-12-09 23:09:54,consumeMessage:rocketmq 发送延迟消息 3,time:2020-12-09 23:09:44
//        2020-12-09 23:09:55,consumeMessage:rocketmq 发送延迟消息 4,time:2020-12-09 23:09:45
//        2020-12-09 23:09:56,consumeMessage:rocketmq 发送延迟消息 5,time:2020-12-09 23:09:46
//        2020-12-09 23:09:57,consumeMessage:rocketmq 发送延迟消息 6,time:2020-12-09 23:09:47
//        2020-12-09 23:09:58,consumeMessage:rocketmq 发送延迟消息 7,time:2020-12-09 23:09:48
//        2020-12-09 23:09:59,consumeMessage:rocketmq 发送延迟消息 8,time:2020-12-09 23:09:49
//        2020-12-09 23:10:01,consumeMessage:rocketmq 发送延迟消息 9,time:2020-12-09 23:09:50



    }
}
